package com.atguigu.state;

import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.optimizer.operators.MapDescriptor;
import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.server.admin.JsonOutputter;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;


/**
 * 算子状态 - 广播状态
 *
 * todo 代码重写二次并测试,已理解
 *
 */

public class Flink04_OperatorBroadcastState {
  public static void main(String[] args) {
      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      env.setParallelism(1);

      //需求：通过广播一个标记，让算子的每个并行实例按照标记执行不同的逻辑代码

      //1.首先创建一天基本数据流
      DataStreamSource<String> ds1 = env.socketTextStream("hadoop102", 8888);

      //2.基于基本流进行广播操作得到广播流
        //对象 mapStateDesc 被创建，用于描述广播状态的键值对类型。
      MapStateDescriptor<String, String> mapStateDesc = new MapStateDescriptor<>("mapStateDesc", Types.STRING, Types.STRING);
      BroadcastStream<String> broadcastDs = env.socketTextStream("hadoop102", 9999)
              .broadcast(mapStateDesc);//括号需要传参，描述广播状态，去上面new一个


      //connect
      ds1.connect(broadcastDs)
              .process(

                      new BroadcastProcessFunction<String, String, String>() {
                       /**
                       * 处理数据流中的数据
                       *
                       * 从广播状态中获取状态数据，结合广播状态，对基本数据流进行相应的逻辑处理
                       *
                       * 读取广播状态，执行相应逻辑操作
                       */

                          @Override
                          public void processElement(String key, BroadcastProcessFunction<String, String, String>.ReadOnlyContext readOnlyContext, Collector<String> out) throws Exception {
                              //获取广播状态,这里只能读，只是获取，不能操作广播状态
                              ReadOnlyBroadcastState<String, String> broadcastState = readOnlyContext.getBroadcastState(mapStateDesc);
                              //依据广播状态的key获得相应的值
                              String flag = broadcastState.get("flag");
                              //依据广播状态的值，执行相应的逻辑

                              if ("1".equals(flag)){
                                  System.out.println("执行逻辑1。。。。。");
                              } else if ("2".equals(flag)) {
                                  System.out.println("执行逻辑2。。。。。");
                              } else if ("3".equals(flag)) {
                                   System.out.println("执行逻辑3。。。。。");
                              }else {
                                   System.out.println("执行默认逻辑");
                              }
                              out.collect(key);//这里输出啥
                          }
                       /**
                       * 处理广播流中的数据
                       *
                       * 获取到广播流中的数据，更新广播状态
                       *
                       * 对广播状态进行增删改查
                       */

                          @Override
                          public void processBroadcastElement(String key, BroadcastProcessFunction<String, String, String>.Context context, Collector<String> out) throws Exception {
                              BroadcastState<String, String> broadcastState = context.getBroadcastState(mapStateDesc);
                              //更新广播状态
                              broadcastState.put("flag",key);

                          }
                      }



              )
              .setParallelism(2)
              .print();





      try {
          env.execute();
      } catch (Exception e) {
          throw new RuntimeException(e);
      }
  }
}


